package com.zbj.storm.old;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * ExclamationBolt
 *
 * @author weigang
 * @create 2019-09-02
 **/
public class ExclamationBolt extends BaseRichBolt {

    private Logger logger = LoggerFactory.getLogger(ExclamationBolt.class);

    private OutputCollector collector;

    /**
     * prepare()为Bolt提供了一个OutputCollector用于Bolt中发出元祖 元祖可以随时从 prepare execute cleanup甚至在
     * 另一个线程中异步发出 当前prepare实现只是将OutputCollector作为实例变量保存 以便稍后在execute方法中使用
     *
     * @param topoConf
     * @param context
     * @param collector
     */
    @Override
    public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    /**
     * execute()从一个Bolt的输入接收一个元组 此execute取数组的第一个字段并发出追加字符串"levelUp!"得到字符串
     * 如果您实现了一个订阅多个输入源的Bolt 您可以通过Tuple#getSourceComponent() 找出Tuple来自哪个组件
     *
     * @param input
     */
    @Override
    public void execute(Tuple input) {
        String sourceComponent = input.getSourceComponent();
        // 输入元组作为第一个参数传递emit
        collector.emit(input, new Values(input.getString(0) + "levelUp!"));
        //System.out.println(input.getString(0));
        logger.info(input.getString(0));
        // 输入元组在最后一行被激活 这些是Storm的可靠性API的一部分 用于保证不会丢失数据
        collector.ack(input);

    }

    /**
     * 声明Bolt发出1元组 其中一个字段称为"word"
     *
     * @param declarer
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word"));
    }

    /**
     * 如果 implements IRichBolt 还需要重写两个方法
     * 当Bolt被关闭时 调用cleanup方法 并且应该清楚所有打开的资源
     * 无法保证在集群上调用此方法 例如 如果任务正在运行的计算机爆炸 则无法调用该方法
     */
    @Override
    public void cleanup() {
        super.cleanup();
    }

    /**
     * 允许配置此组件运行方式的各个方面
     *
     * @return
     */
    @Override
    public Map<String, Object> getComponentConfiguration() {
        return super.getComponentConfiguration();
    }
    // 但是一般情况下我们不需要这两个方法，所以我们可以通过继承BaseRichBolt来定义Bolt
}